package org.example.realtime.traffic.dwd.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.realtime.jtp.common.utils.JdbcUtil;
import org.example.realtime.jtp.common.utils.KafkaUtil;


/**
 * @Title: trafficrealtimeWeifadwdJob
 * @Author Lianzy
 * @Package org.example.realtime.traffic.dwd.log.job
 * @Date 2025/5/27 10:10
 * @description 实时获取违法事件日志
 */
public class trafficrealtimeWeifadwdJob {
    public static void main(String[] args) throws Exception {

        //执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.数据源
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "traffic_events");
//        kafkaDataStream.print();
        /**
         * {
         *   "eventId": "fde7ce20-73e9-43de-865f-4272fe1e7493",
         *   "cameraId": "CAM-082",
         *   "licensePlate": "黑F-10565",
         *   "vehicleType": "小型汽车",
         *   "roadId": "ROAD-014",
         *   "crossingId": "CROSS-004",
         *   "ts": 1748309099769,
         *   "speed": 100.3,
         *   "direction": "北",
         *   "lane": 2,
         *   "eventType": "overspeed",
         *   "areaId": "AREA-001"
         * }
         */
        SingleOutputStreamOperator<String> eventType = kafkaDataStream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return !"normal".equals(jsonObject.getString("eventType"));
            }
        });

        eventType.print();
//        SingleOutputStreamOperator<String> jsonStream = eventType.map(new RichMapFunction<String, String>() {
//            @Override
//            public String map(String value) throws Exception {
//                JSONObject jsonObject = new JSONObject();
//                jsonObject.put("eventId", jsonObject.getString("eventId"));
//                jsonObject.put("cameraId", jsonObject.getString("cameraId"));
//                jsonObject.put("licensePlate", jsonObject.getString("licensePlate"));
//                jsonObject.put("vehicleType", jsonObject.getString("vehicleType"));
//                jsonObject.put("roadId", jsonObject.getString("roadId"));
//                jsonObject.put("crossingId", jsonObject.getString("crossingId"));
//                jsonObject.put("ts", jsonObject.getLong("ts"));
//                jsonObject.put("speed", jsonObject.getDouble("speed"));
//                jsonObject.put("direction", jsonObject.getString("direction"));
//                jsonObject.put("lane", jsonObject.getInteger("lane"));
//                jsonObject.put("eventType", jsonObject.getString("eventType"));
//                jsonObject.put("areaId", jsonObject.getString("areaId"));
//                return jsonObject.toJSONString();
//            }
//        });

//        eventStream.print();
        String[] columns = new String[]{"eventId", "cameraId", "licensePlate", "vehicleType", "roadId", "crossingId", "ts", "speed", "direction", "lane", "eventType", "areaId"};
        String upsertSql ="INSERT INTO\n" +
                "traffic_log.traffic_violations\n" +
                "(eventId, cameraId, licensePlate, vehicleType, roadId, crossingId, ts, speed, direction, lane, eventType, areaId)\n" +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
                JdbcUtil.sinkToClickhouseUpsertJson(eventType, columns,upsertSql );

        env.execute();
    }
}
